package com.nepu.spark.realtime.app


import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.nepu.spark.realtime.bean.{PageActiveLog, PageDisplayLog, PageLog, StartLog}
import com.nepu.spark.realtime.constant.KafkaConstant
import com.nepu.spark.realtime.util.{KafkaOffsetUtils, MyKafkaUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 本类的主要的目的是对ODS层的日志数据进行分流
  *
  * @author chenshuaijun
  * @create 2022-10-29 19:29
  */

/**
  * 步骤：
  * 1、创建SparkStreaming上下文环境
  *
  * 2、数据分流
  *   2.1、解析数据
  *   2.2、将对应的数据发送到对应的主题中
  *
  */
object OdsLogDataDiversion {


  def main(args: Array[String]): Unit = {
    // 1、我们在构建上下文环境的时候，有一个点是需要注意的，我们使用setMaster("local[4]")设置并行度的时候，
    // 一定要和kafka中的分区的个数设置成一样
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("ods_log_data_diversion")
    val ssc = new StreamingContext(conf, Seconds(3))

    // topic
    val topic: String = KafkaConstant.ODS_BASE_LOG_TOPIC
    // 消费者组
    val groupId: String = "ods_base_log"
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    // 先从redis中获取offset
    val offset: Map[TopicPartition, Long] = KafkaOffsetUtils.getOffset(topic, groupId)
    if (offset != null && offset.nonEmpty) {
      // 如果Redis中维护的有offset就使用自己管理的offset
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId, offset)
    } else {
      // 如果redis中没有offset就使用,kafka管理的
      kafkaDStream = MyKafkaUtils.getKafkaDStream(ssc, topic, groupId)
    }
    // 获取本次消费的offset
    var offsetRange: Array[OffsetRange] = null
    val offsetRangeDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        val offsetRanges: HasOffsetRanges = rdd.asInstanceOf[HasOffsetRanges]
        offsetRange = offsetRanges.offsetRanges
        rdd
      }
    )
    // 将消费出来的数据进行结构的转换，转换成jsonObject
    val jsonObjDStream: DStream[JSONObject] = offsetRangeDStream.map(
      consumerRecord => {
        val record: String = consumerRecord.value()
        val jSONObject: JSONObject = JSON.parseObject(record)
        jSONObject
      }
    )


    // 3、开始对数据进行分流
    // 3.1 定义分流所使用到的topic
    // 启动日志的主题
    val dwd_start_log: String = KafkaConstant.DWD_START_LOG
    // 错误数据的主题
    val dwd_error_log: String = KafkaConstant.DWD_ERROR_INFO
    // 页面日志
    val dwd_page_log: String = KafkaConstant.DWD_PAGE_LOG
    // 动作数据
    val dwd_page_action: String = KafkaConstant.DWD_PAGE_ACTION
    // 曝光的日志
    val dwd_page_display: String = KafkaConstant.DWD_PAGE_DISPLAY

    // 开始分流数据
    jsonObjDStream.foreachRDD(
      jsonObjRDD => {

        jsonObjRDD.foreachPartition(
          jsonObjIter => {
            for (jsonObj <- jsonObjIter) {
              // 首先是对错误数据的处理,错误数据是不需要进行解析的，直接丢到错误日志的topic中
              val errLog: String = jsonObj.getString("err")
              if (errLog != null) {
                MyKafkaUtils.send(dwd_error_log, jsonObj.toJSONString)
              }
              // 解析公共字段
              val commonObj: JSONObject = jsonObj.getJSONObject("common")
              val ar: String = commonObj.getString("ar") // 获得省份id
              val userId: String = commonObj.getString("uid") // 用户Id
              val os: String = commonObj.getString("os") // 操作系统
              val ch: String = commonObj.getString("ch") // 渠道
              val is_new: String = commonObj.getString("is_new") // 是否是首次登录
              val md: String = commonObj.getString("md")
              val mid: String = commonObj.getString("mid") // 设备id
              val vc: String = commonObj.getString("vc")
              val ba: String = commonObj.getString("ba") // 品牌
              val ts: Long = jsonObj.getLong("ts") // 获取时间戳

              // 提取页面数据
              val pageObj: JSONObject = jsonObj.getJSONObject("page")
              if (pageObj != null) {
                // 解析页面的属性
                val page_id: String = pageObj.getString("page_id")
                val item: String = pageObj.getString("item")
                val duringTime: Long = pageObj.getLong("during_time")
                val itemType: String = pageObj.getString("item_type")
                val lastPageId: String = pageObj.getString("last_page_id")
                val sourceType: String = pageObj.getString("source_type")
                // 发送页面数据
                val pageLog = PageLog(ar, userId, os, ch, is_new, md, mid, vc, ba, ts, page_id, item, duringTime, itemType, lastPageId, sourceType)
                MyKafkaUtils.send(dwd_page_log, JSON.toJSONString(pageLog, new SerializeConfig(true)))

                // 因为只有有了页面数据才有曝光数据，下面进行曝光数据的解析
                val displaysJsonArr: JSONArray = jsonObj.getJSONArray("displays")
                // 需要先判断一下能不能得到曝光的数据
                if (displaysJsonArr != null && displaysJsonArr.size() > 0) {
                  for (i <- 1 until displaysJsonArr.size()) {
                    val display: JSONObject = displaysJsonArr.getJSONObject(i)
                    val displayType: String = display.getString("display_type")
                    val disPlayItem: String = display.getString("item")
                    val displayItemType: String = display.getString("item_type")
                    val displayPosId: String = display.getString("pos_id")
                    val displayOrder: String = display.getString("order")
                    val displayPageLog = PageDisplayLog(ar, userId, os, ch, is_new, md, mid, vc, ba, ts, displayType, disPlayItem, displayItemType, displayPosId, displayOrder)
                    MyKafkaUtils.send(dwd_page_display, JSON.toJSONString(displayPageLog, new SerializeConfig(true)))
                  }
                }

                // 对动作数据进行提取
                val actionJsonArr: JSONArray = jsonObj.getJSONArray("actions")
                if (actionJsonArr != null && actionJsonArr.size() > 0) {
                  for (i <- 1 until actionJsonArr.size()) {
                    val action: JSONObject = actionJsonArr.getJSONObject(i)
                    val actionItem: String = action.getString("item")
                    val actionId: String = action.getString("action_id")
                    val actionItemType: String = action.getString("item_type")
                    val actionTs: Long = action.getLong("ts")
                    val pageActionLog = PageActiveLog(ar, userId, os, ch, is_new, md, mid, vc, ba, ts, actionItem, actionId, actionItemType, actionTs)
                    MyKafkaUtils.send(dwd_page_action, JSON.toJSONString(pageActionLog, new SerializeConfig(true)))
                  }
                }

              }

              // 提取启动数据
              val startJsonObj: JSONObject = jsonObj.getJSONObject("start")
              if (startJsonObj != null) {
                val startEntry: String = startJsonObj.getString("entry")
                val startOpenAdSkipMs: Long = startJsonObj.getLong("open_ad_skip_ms")
                val startOpenAdMs: Long = startJsonObj.getLong("open_ad_ms")
                val startLoadingTime: Long = startJsonObj.getLong("loading_time")
                val startOpenAdId: Long = startJsonObj.getLong("open_ad_id")
                val startLog = StartLog(ar, userId, os, ch, is_new, md, mid, vc, ba, ts, startEntry, startOpenAdSkipMs, startOpenAdMs, startLoadingTime, startOpenAdId)
                MyKafkaUtils.send(dwd_start_log, JSON.toJSONString(startLog, new SerializeConfig(true)))
              }
            }
            // 刷写kafka中缓存的数据到分区磁盘，我们对于数据的刷写是不可以放在driver端的，
            // 因为原本的kafka的producer是在Executor端创建的，如果我们在driver端进行刷写，那么算写的对象是有问题的。
            MyKafkaUtils.flush()
          }
        )
        // 在这里手动的提交offset
        KafkaOffsetUtils.saveOffset(topic, groupId, offsetRange)
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }

}
